/* Copyright (c) 2023 Renmin University of China
RMDB is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
        http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

#pragma once

#include "errors.h"
#include "execution_defs.h"
#include "execution_manager.h"
#include "executor_abstract.h"
#include "record/merge_sort.h"
#include "index/ix.h"
#include "system/sm.h"
#include "system/sm_meta.h"
#include <functional>

class MergeJoinExecutor : public AbstractExecutor {

  private:
    std::unique_ptr<AbstractExecutor> left_;            
    std::unique_ptr<AbstractExecutor> right_;           
    size_t len_;                                        
    std::vector<ColMeta> cols_;                         
    std::vector<Condition> fed_conds_; 
    bool is_end_ = false;
    bool use_index_;  
    ColMeta left_col_, right_col_;   
    std::unique_ptr<RmRecord> current_tuple_;                         
                                                              
    std::vector<std::unique_ptr<RmRecord>> records_;

    enum { LEFT = 0, RIGHT = 1 };                                  

  public:

    MergeJoinExecutor(std::unique_ptr<AbstractExecutor> left, std::unique_ptr<AbstractExecutor> right,
                    std::vector<Condition> conds, bool use_index = false)
        : left_(std::move(left)), right_(std::move(right)), fed_conds_(std::move(conds)), use_index_(use_index) {
        
        initialize_records();
        initialize_columns();
        initialize_join_columns();
    }

    void beginTuple() override {
        openSortOutputFiles();
        generateHeaders();

        if (use_index_) {
            initializeExecutorsWithIndex();
        } else {
            sleep(2);
            initializeExecutorsWithoutIndex();
        }
        nextTuple();
    }

    void nextTuple() override {
        if (checkEndCondition()) {
            is_end_ = true;
            mergeLR();
            return;
        }

        readRecords();
        generateRecords();

        int result = compareRecords();

        while (result != 0 && !checkEndCondition()) {
            processMismatchedRecords(result);
            result = compareRecords();
        }

        if (result != 0) {
            is_end_ = true;
            return;
        }

        updateBufferWithJoinedTuple();
    }

    [[nodiscard]] bool is_end() const override {
        return is_end_;
    };

    std::unique_ptr<RmRecord> Next() override {
        return std::move(current_tuple_);
    };

    Rid &rid() override {
        return _abstract_rid;
    }

    [[nodiscard]] const std::vector<ColMeta> &cols() const override {
        return cols_;
    };

    [[nodiscard]] size_t tupleLen() const override {
        return len_;
    };

    ColMeta get_col_offset(const TabCol &target) override {
        return *get_col(cols_, target);
    }
private:
// Private helper functions
    std::ofstream L_result, R_result;
    void openSortOutputFiles() {
        L_result.open("sorted_resultsL.txt");
        R_result.open("sorted_resultsR.txt");
        if (R_result.fail() || L_result.fail()) {
            throw InternalError("openSortOutputFiles Error");
        }
    }

    void generateHeaders() {
        generateHead(left_->cols(), L_result);
        generateHead(right_->cols(), R_result);
    }

    void initializeExecutorsWithIndex() {
        left_->beginTuple();
        right_->beginTuple();
    }

    MergeSort L_sort, R_sort;
    void initializeExecutorsWithoutIndex() {
        L_sort = startSort(left_, left_col_);
        R_sort = startSort(right_, right_col_);
    }

    bool checkEndCondition() {
        bool end_when_use_index = use_index_ && (left_->is_end() || right_->is_end());
        bool end_when_not_use_index = !use_index_ && (!(L_sort.getLen() && R_sort.getLen()));
        return end_when_use_index || end_when_not_use_index;
    }

    void readRecords() {
        getRecord(LEFT);
        getRecord(RIGHT);
    }

    void generateRecords() {
        generateRecord(left_->cols(), L_result, records_[0]->data);
        generateRecord(right_->cols(), R_result, records_[1]->data);
    }

    int compareRecords() {
        auto lvalue = convertValue(static_cast<char *>(records_[0]->data), left_col_);
        auto rvalue = convertValue(static_cast<char *>(records_[1]->data), right_col_);
        return (lvalue < rvalue) ? -1 : (lvalue > rvalue) ? 1 : 0;
    }

    void processMismatchedRecords(int &result) {
        if (result < 0) {
            getRecord(LEFT);
            generateRecord(left_->cols(), L_result, records_[0]->data);
        } else {
            getRecord(RIGHT);
            generateRecord(right_->cols(), R_result, records_[1]->data);
        }
    }

    void updateBufferWithJoinedTuple() {
        assert(current_tuple_ == nullptr);
        current_tuple_ = std::make_unique<RmRecord>(len_);
        memcpy(current_tuple_->data, records_[0]->data, left_->tupleLen());
        memcpy(current_tuple_->data + left_->tupleLen(), records_[1]->data, right_->tupleLen());
    }

    void initialize_join_columns() {
        for (const auto &cond : fed_conds_) {
            if (cond.is_rhs_val || cond.op != OP_EQ) {
                continue;
            }

            bool is_left_to_right = cond.lhs_col.tab_name == left_->tableName() && cond.rhs_col.tab_name == right_->tableName();
            bool is_right_to_left = cond.lhs_col.tab_name == right_->tableName() && cond.rhs_col.tab_name == left_->tableName();

            if (is_left_to_right || is_right_to_left) {
                left_col_ = *get_col(is_left_to_right ? left_->cols() : right_->cols(), is_left_to_right ? cond.lhs_col : cond.rhs_col);
                right_col_ = *get_col(is_left_to_right ? right_->cols() : left_->cols(), is_left_to_right ? cond.rhs_col : cond.lhs_col);
            }
        }
    }


    void initialize_columns() {
        len_ = left_->tupleLen() + right_->tupleLen();
        cols_ = left_->cols();
        auto right_cols = right_->cols();
        for (auto &col : right_cols) {
            col.offset += left_->tupleLen();
        }
        cols_.insert(cols_.end(), right_cols.begin(), right_cols.end());
    }

    void initialize_records() {
        records_.emplace_back(std::make_unique<RmRecord>(left_->tupleLen()));
        records_.emplace_back(std::make_unique<RmRecord>(right_->tupleLen()));
    }

    void mergeLR() {
        if (use_index_) {
            generateRecordWithIndex();
        } else {
            generateRecordWithoutIndex();
        }

        L_result.close();

        {
            std::ifstream file("sorted_resultsL.txt", std::ios::binary | std::ios::in);
            R_result << file.rdbuf();
        }

        R_result.close();
        unlink("sorted_resultsL.txt");
        rename("sorted_resultsR.txt", "sorted_results.txt");
    }


    void generateRecordWithIndex() {
        while (!left_->is_end()) {
            getRecord(LEFT);
            if (!left_->is_end()) break;
            generateRecord(left_->cols(), L_result, records_[0]->data);
        }
        while (!right_->is_end()) {
            getRecord(RIGHT);
            if (!right_->is_end()) break;
            generateRecord(right_->cols(), R_result, records_[1]->data);
        }
    }

    void generateRecordWithoutIndex() {
        while (L_sort.getLen()) {
            getRecord(LEFT);
            generateRecord(left_->cols(), L_result, records_[0]->data);
        }
        while (R_sort.getLen()) {
            getRecord(RIGHT);
            generateRecord(right_->cols(), R_result, records_[1]->data);
        }
    }

    MergeSort startSort(std::unique_ptr<AbstractExecutor> &executor, const ColMeta &joined_col) {

        MergeSort sorter(executor->tupleLen(), const_cast<void *>(static_cast<const void *>(&joined_col)));

        for (executor->beginTuple(); !executor->is_end(); executor->nextTuple()) {
            sorter.beginSort(executor->Next()->data);
        }

        sorter.endSort();
        return sorter;
    }


    void generateHead(const std::vector<ColMeta> &cols, std::ofstream &output) {
        output << "|";
        for (const auto &col : cols) {
            output << " " << (!col.alias.empty() ? col.alias : col.name) << " |";
        }
        output << "\n";
    }

    void generateRecord(const std::vector<ColMeta> &cols, std::ofstream &output, const char *record) {
        output << "|";
        for (const auto &col : cols) {
            const char *rec_buf = record + col.offset;
            std::string col_str;
            switch (col.type) {
                case TYPE_INT:
                    col_str = std::to_string(*reinterpret_cast<const int *>(rec_buf));
                    break;
                case TYPE_FLOAT:
                    col_str = std::to_string(*reinterpret_cast<const float *>(rec_buf));
                    break;
                case TYPE_STRING:
                    col_str = std::string(rec_buf, col.len).c_str();
                    break;
            }
            output << " " << col_str << " |";
        }
        output << "\n";
    }

    void getRecord(int record_idx) {
        if (use_index_) {
            auto executor = record_idx == 0 ? left_.get() : right_.get();
            records_[record_idx] = executor->Next();
            executor->nextTuple();
        } else {
            switch (record_idx) {
                case LEFT:
                    L_sort.nextRcd(records_[record_idx]->data);
                    break;
                case RIGHT:
                    R_sort.nextRcd(records_[record_idx]->data);
                    break;
                default:
                    throw InternalError("getRecord Error");
            }
        }
    }

};